/**
 * MessageSender.java<br>
 *
 * @author aaronwong.sd@foxmail.com
 * @version 下午4:20:39
 */
package com.iwechat.service.runner;

import java.util.concurrent.ConcurrentLinkedQueue;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import com.iwechat.mybatis.generator.model.TemplateMessage;
import com.iwechat.service.TempMsgPriorityQueue;
import com.iwechat.util.PropertyUtils;

/**
 * 消息发送器
 *
 * @author aaronwong.sd@foxmail.com
 */
@Service
public class MessageSender implements Runnable {
    private static final Logger                                logger            = LogManager.getLogger();
    private static final long                                  _1S               = 1000;
    /** 多线程及多模块共享的的 */
    public static final ConcurrentLinkedQueue<TemplateMessage> passedMessagePool = new ConcurrentLinkedQueue<TemplateMessage>();

    private volatile boolean                                   run               = true;

    @Resource(name = "RedisPriorityQuenue")
    TempMsgPriorityQueue                                       queue;
    @Autowired
    ThreadPoolTaskExecutor                                     executor;

    /*
     * (non-Javadoc)
     *
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        String url = "";
        // 延迟,等待对象初始化
        try {
            synchronized (this) {
                this.wait(_1S);
            }
            url = PropertyUtils.getInstance().getStringValue("httpclient", "wxAdapter.url");
        } catch (Exception e) {
        }
        while (run) {
            try {
                logger.info("message queue.take()...");
                TemplateMessage tm = queue.take();
                // 有消息发送
                if (tm != null) {
                    TmHttpRunner tmr = new TmHttpRunner(url, tm);
                    // 多线程发送批量更新
                    try {
                        executor.execute(tmr);
                    } catch (TaskRejectedException e) {
                        logger.error(e, e);
                        // 满负荷,停止10s发送
                        synchronized (this) {
                            this.wait(10 * _1S);
                        }
                        // 重试一次
                        try {
                            executor.execute(tmr);
                        } catch (TaskRejectedException e2) {
                            logger.error(e, e);
                        }
                    }
                }
            } catch (Exception ex) {
                logger.error(ex, ex);
            }
        }
    }

    @PostConstruct
    public void init() {
        new Thread(this, "MessageSender-Main").start();

        logger.info("### " + this + " init succeed!");
    }

    @PreDestroy
    public void destory() {
        this.run = false;
        synchronized (this) {
            this.notifyAll();
        }
        logger.info("### " + this + " destory succeed!");
    }
}
